package org.apache.rocketmq.client.codecoord.util;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import org.apache.rocketmq.client.codecoord.RocketMqConstant;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.List;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * 消息发送帮助器
 *
 * @author tianxincoord@163.com
 * @since 2022/4/21
 */
public class ProducerHelper {

    /**
     * 发送同步消息
     */
    public static void sendSyncMessage(Supplier<Message> consumer) throws Exception {
        sendMessage((producer) -> {
            Message message;
            try {
                message = consumer.get();
                return producer.send(message);
            } catch (Exception e) {
                SendResult sendResult = new SendResult();
                sendResult.setOffsetMsgId(e.getMessage());
                return sendResult;
            }
        });
    }

    /**
     * 发送同步消息
     */
    public static void sendBatchMessage(Supplier<List<Message>> consumer) throws Exception {
        sendMessage((producer) -> {
            List<Message> messageList;
            try {
                messageList = consumer.get();
                return producer.send(messageList);
            } catch (Exception e) {
                SendResult sendResult = new SendResult();
                sendResult.setOffsetMsgId(e.getMessage());
                return sendResult;
            }
        });
    }

    /**
     * 发送延迟步消息
     */
    public static void sendDelayMessage(Supplier<Message> consumer, int delayLevel) throws Exception {
        sendMessage((producer) -> {
            Message message;
            try {
                message = consumer.get();
                // 指定消息延迟等级
                message.setDelayTimeLevel(delayLevel);
                return producer.send(message);
            } catch (Exception e) {
                SendResult sendResult = new SendResult();
                sendResult.setOffsetMsgId(e.getMessage());
                return sendResult;
            }
        });
    }

    /**
     * 发送顺序消息
     */
    public static void sendOrderMessage(Supplier<List<Message>> consumer) throws Exception {
        sendMessage((producer) -> {
            List<Message> messageList;
            try {
                messageList = consumer.get();
                for (Message message : messageList) {
                    producer.send(message, new MessageQueueSelector() {
                        @Override
                        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                            // 顺序消息使用业务id作为主键(根据业务选择则)
                            long index = Long.parseLong(arg.toString()) % mqs.size();
                            return mqs.get((int) index);
                        }
                    }, message.getKeys());
                }
                return new SendResult();
            } catch (Exception e) {
                SendResult sendResult = new SendResult();
                sendResult.setOffsetMsgId(e.getMessage());
                e.printStackTrace();
                return sendResult;
            }
        });
    }

    /**
     * 消息发送
     */
    public static void sendMessage(Function<DefaultMQProducer, SendResult> consumer) throws Exception {
        // 处理控制台报错
        // org.apache.rocketmq.client.log.ClientLogger
        System.setProperty("rocketmq.client.logUseSlf4j", "true");
        DefaultMQProducer producer = new DefaultMQProducer(RocketMqConstant.GROUP);
        producer.setNamesrvAddr(RocketMqConstant.NAME_SRV_ADDRESS);
        producer.start();
        SendResult sendResult = consumer.apply(producer);
        System.out.printf("消息发送成功[%s]\n", JSONObject.toJSONString(sendResult, SerializerFeature.PrettyFormat));
        producer.shutdown();
    }
}
